package com.shujia.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo01WordCountOnStreaming {
  def main(args: Array[String]): Unit = {

    // 创建StreamingContext

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo01WordCountOnStreaming")
    conf.setMaster("local[2]") // 在ssc中需要一个线程专门用于接收数据，所以至少需要两个线程才能完成数据处理

    val sc: SparkContext = new SparkContext(conf)

    // Durations 指定每个批次的大小
    // 在这里相当于每5s处理一次，即每5s会产生一个job，如果该job不能在5s内完成，容易造成任务的堆积最终使程序崩溃
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    // 接收数据：通过socket接收数据
    // 需要在虚拟机先启动Server端：nc -lk 8888
    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      .flatMap(line => line.split(","))
      .map(word => (word, 1))
      .reduceByKey(_ + _)
      .print() // ssc 打印数据的方式

    // 启动程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
